package gspushmq.v1.produce.job.sys;

import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.Map;
import java.util.Set;

import gspushmq.v1.consumer.core.MessageQueue;
import org.apache.commons.lang3.StringUtils;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Sets;

import ggframework.mongodb.BasicDBObject;
import ggframework.mongodb.DBCollection;
import ggframework.mongodb.DBCursor;
import ggframework.mongodb.DBObject;
import gspushmq.model.ConsumerWays;
import gspushmq.model.DataMessage;
import gspushmq.model.LinkEnum;
import gspushmq.model.MessageType;
import gspushmq.model.PrefixEnum;
import gspushmq.model.SkipConstant;
import gspushmq.model.UserType;
import gspushmq.util.DateUtil;
import gspushmq.util.GGLogger;
import gspushmq.util.MongoUtil;
import gspushmq.util.SysUtil;
import gspushmq.v1.produce.GGDataStorage;
import gspushmq.v1.produce.job.core.GGServiceJob;

/**
 *
 * @description 直播预约提醒消息生产
 * @param
 * @version 1.0
 * @author chengsja
 * @date 2018/12/20 16:12
 *
 */
public class GGMyOrderVideoJob extends GGServiceJob {

    //直播表
    private static DBCollection videoV3 = null;
    //预定表
    private static DBCollection orderV3 = null;

    @Override
    protected String setJobName() {
        return PrefixEnum.RESERVATION.getMqName();
    }

    @Override
    protected void before() {
        videoV3 = MongoUtil.getFtLiveCollection("cs_live_video");
        orderV3 = MongoUtil.getFtLiveCollection("cas_live_order");
    }

    @Override
    public void doServiceJob(Map<String, Object> context) {
        try {
            info();
        } catch (InterruptedException e) {
            GGLogger.error(GGMyOrderVideoJob.class, "---队列put异常：" + e);
        }
    }

    public void info() throws InterruptedException {
        MessageType messageType = new MessageType();
        messageType.setMq_type(PrefixEnum.RESERVATION.getMqType());
        messageType.setMq_sub_type(PrefixEnum.RESERVATION.getMqSubType());

        DBObject query = new BasicDBObject();
        query.put("status", 1);
        query.put("live_status", 0);
        DBObject timeQuery = new BasicDBObject("$gt", new Date());
        timeQuery.append("$lt", DateUtil.getAddMinute(new Date(), 15));
        query.put("live_time_start", timeQuery);

        BasicDBObject fields = new BasicDBObject();
        fields.put("_id", 0);
        fields.put("live_time_start", 1);
        fields.put("live_time_end", 1);
        fields.put("live_title", 1);
        fields.put("uuid", 1);
        fields.put("introduction",1);
        DBObject sort = new BasicDBObject("update_time", -1);
        //判读是否有快开始的直播
        if (videoV3.count(query) <= 0) {
            return;
        }
        //过滤已生产的直播
        Set<String> existsUUIDSet = SysUtil.filterAlreadyExistsUUid(messageType);
        GGLogger.info(getClass(), "-----需要过滤已生产直播预约：" + JSON.toJSONString(existsUUIDSet));
        if (existsUUIDSet != null && existsUUIDSet.size() > 0) {
            //已经生产的预约消息
            query.put("uuid", new BasicDBObject("$nin", existsUUIDSet));
        }
        DBCursor cursor = videoV3.find(query, fields).sort(sort);
        //直播uuid池
        Set<String> orderUUidSet = Sets.newHashSet();
        //临时消息池
        Set<DataMessage> dataMessageSet = Sets.newHashSet();
        while (cursor.hasNext()) {
            DBObject o = cursor.next();
            Date startTime = o.getDate("live_time_start");
            String uuid = o.getString("uuid");
            //生产消息
            DataMessage dm = new DataMessage();
            dm.setUuid(uuid);
            dm.setMqType(PrefixEnum.RESERVATION.getMqType());
            dm.setMqSubType(PrefixEnum.RESERVATION.getMqSubType());
            dm.setUserType(UserType.ASSIGN);
            //简介
            String content = o.getString("introduction");
            dm.setContent(content);
            dm.setTitle("【"+o.get("live_title")+"】即将开始!");
            dm.setExpireDate(startTime);
            dm.setStartDate(new Date());
            //账号id推送
            dm.setCrowdType(ConsumerWays.ALIAS.getWay());
            orderUUidSet.add(uuid);
            //add
            dataMessageSet.add(dm);
        }
        cursor.close();
        GGLogger.info(getClass(),"-----过滤后生产的直播预约："+ JSON.toJSONString(orderUUidSet));
        if (dataMessageSet.size() == 0){
            //没有直播消息就停止
            return;
        }
        //查询预约直播用户
        ArrayListMultimap<String, Long> uuidWithUidsMulMap = ArrayListMultimap.create();
        query = new BasicDBObject();
        query.append("uuid", new BasicDBObject("$in", orderUUidSet));
        query.append("order_status", new BasicDBObject("$in", Arrays.asList(1, 2)));
        fields = new BasicDBObject();
        fields.append("_id", 0);
        fields.append("user_id", 1);
        fields.append("uuid", 1);
        fields.append("source", 1);
        cursor = orderV3.find(query, fields);
        while (cursor.hasNext()) {
            DBObject o = cursor.next();
            uuidWithUidsMulMap.put(o.getString("uuid"), o.getLong("user_id"));
        }
        cursor.close();
        if (uuidWithUidsMulMap.asMap().size() == 0) {
            //没有预约的用户就停止
            return;
        }
        //组装dm
        for (DataMessage dataMessage : dataMessageSet) {
            //单条直播消息
            String uuid = dataMessage.getUuid();
            if (StringUtils.isBlank(uuid)) {
                continue;
            }
            Collection<Long> uidList = uuidWithUidsMulMap.asMap().get(uuid);
            //过滤重复账号
            Set<Long> uidSet = Sets.newHashSet(uidList);
            if (uidList != null && uidList.size() > 0) {
                //置入订阅用户id字符串
                dataMessage.setCrowdValue(StringUtils.join(uidSet, ";"));
                //内链，兼容wxd和lyjx页面
                dataMessage.setLinkType(LinkEnum.INNER.getLinkType());
                dataMessage.setTarget(uuid);
            }
            //开始复制消息到对应的产品线
            for (Integer productLine : Sets.newHashSet(4, 28, 901) ) {
                DataMessage d1 = deepClone(dataMessage);
                //置入产品线
                d1.setProductLine(productLine);
                //url 王旭东需要
                d1.setLinkUrl(getLinkUrl(productLine)+d1.getTarget());
                //module: 刘永建新需要
                d1.setModule(getModule(productLine));
                //加入总消息池
                MessageQueue.messageQueue.add(d1);
                //入库
                SysUtil.sendToMongo(d1);
                //日志
                SysUtil.logPattern(GGMyOrderVideoJob.class, d1);
            }
        }

    }
/**
 *
 * @description 复制对象
 * @param
 * @version 1.0
 * @author chengsja
 * @date 2018/12/18 10:21
 *
 */
    private DataMessage deepClone(DataMessage dataMessage){
        DataMessage d1 = new DataMessage();
        d1.setModule(dataMessage.getModule());
        d1.setTarget(dataMessage.getTarget());
        d1.setLinkType(dataMessage.getLinkType());
        d1.setProductLine(dataMessage.getProductLine());
        d1.setTitle(dataMessage.getTitle());
        d1.setUuid(dataMessage.getUuid());
        d1.setContent(dataMessage.getContent());
        d1.setCrowdType(dataMessage.getCrowdType());
        d1.setCrowdValue(dataMessage.getCrowdValue());
        d1.setMqType(dataMessage.getMqType());
        d1.setMqSubType(dataMessage.getMqSubType());
        d1.setExpireDate(dataMessage.getExpireDate());
        d1.setStartDate(dataMessage.getStartDate());
        d1.setUserType(dataMessage.getUserType());
        d1.setLinkUrl(dataMessage.getLinkUrl());
        return d1;
    }

    /**
     * 获取不同产品线的不同跳转module
     * Created by chengsj on 2018/11/28
     */
    private String getModule(Integer productLine) {
        if (productLine == null) {
            return null;
        }
        if (productLine == 4) {
            return SkipConstant.LIVE_RESERVATION_4_MODULE;
        } else if (productLine == 28) {
            return SkipConstant.LIVE_RESERVATION_28_MODULE;
        } else if (productLine == 901) {
            return SkipConstant.LIVE_RESERVATION_901_MODULE;
        }
        return null;
    }
    private String getLinkUrl(Integer productLine) {
        if (productLine == null) {
            return null;
        }
        if (productLine == 4) {
            return SkipConstant.LIVE_RESERVATION_4_URL;
        } else if (productLine == 28) {
            return SkipConstant.LIVE_RESERVATION_28_URL;
        } else if (productLine == 901) {
            return SkipConstant.LIVE_RESERVATION_901_URL;
        }
        return null;
    }

}
